/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.computing.commons.helper;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.ipig.computing.commons.pojo.kafka.KafkaRecordInfo;
import org.ipig.computing.commons.pojo.kafka.RecordInfo;
import org.ipig.constants.SymbolCnst;

/**
 * KafkaHelper
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @since 1.0
 */
@Slf4j
public class KafkaHelper {

    /**
     * 将ConsumerRecord<String, String>记录信息转化成KafkaRecordInfo<String, RecordInfo>记录信息
     *
     * @param recordLine ConsumerRecord<String, String>
     * @return KafkaRecordInfo<String, RecordInfo>
     */
    public static KafkaRecordInfo<String, RecordInfo<JSONObject, JSONObject>> toKafkaRecordInfo(final ConsumerRecord<String, String> recordLine) {
        KafkaRecordInfo<String, RecordInfo<JSONObject, JSONObject>> recInfo = new KafkaRecordInfo<String, RecordInfo<JSONObject, JSONObject>>();
        recInfo.setChecksum(recordLine.checksum());
        recInfo.setOffset(recordLine.offset());
        recInfo.setPartition(recordLine.partition());
        recInfo.setTopic(recordLine.topic());
        recInfo.setTimestamp(recordLine.timestamp());
        recInfo.setKey(recordLine.key());
        recInfo.setSerializedKeySize(recordLine.serializedKeySize());
        recInfo.setSerializedValueSize(recordLine.serializedValueSize());
        String textValue = recordLine.value();
        recInfo.setOrigRecord(textValue);
        if (StringUtils.isNotBlank(textValue)) {
            String msg = textValue;
//            System.out.println("源数据："+textValue);
            if (!JSONObject.isValid(textValue)) {
                msg = StringUtils.substring(textValue, 0, textValue.length() - 1);
//                System.out.println("转化后："+msg);
                if (!JSONObject.isValid(msg)) {
                    System.err.println("源数据无法解析 " + textValue);
                    return null;
                }
            }
            RecordInfo<JSONObject, JSONObject> sri = new RecordInfo<JSONObject, JSONObject>();
            if (StringUtils.isNotBlank(msg)) {
                JSONObject jsonObj = JSONObject.parseObject(msg);
                if (jsonObj != null) {
                    JSONObject beforeObject = jsonObj.getJSONObject(RecordInfo.BEFORE);
                    JSONObject afterObject = jsonObj.getJSONObject(RecordInfo.AFTER);
                    //对于OGG过来的数据格式为schema.table
                    String allTableName = jsonObj.getString(RecordInfo.TABLE);
                    JSONArray jsonArray = jsonObj.getJSONArray(RecordInfo.PRIMARY_KEYS);
                    if (afterObject != null) {
                        sri.setAfterData(afterObject);
                    }
                    if (beforeObject != null) {
                        sri.setBeforeData(beforeObject);
                    }
                    sri.setCurrentTs(jsonObj.getString(RecordInfo.CURRENT_TS));
                    sri.setOpTs(jsonObj.getString(RecordInfo.OP_TS));
                    sri.setOpType(jsonObj.getString(RecordInfo.OP_TYPE));
                    sri.setPos(jsonObj.getString(RecordInfo.POS));

                    int index = StringUtils.indexOf(allTableName, SymbolCnst.DOT);
                    if (index >= 0) {
                        sri.setSchema(StringUtils.substringBefore(allTableName, SymbolCnst.DOT));
                        sri.setCode(StringUtils.substringAfter(allTableName, SymbolCnst.DOT));
                    } else {
                        sri.setSchema(StringUtils.EMPTY);
                        sri.setCode(allTableName);
                    }
                    if (jsonArray != null) {
                        String[] arry = new String[jsonArray.size()];
                        jsonArray.toArray(arry);
                        sri.setPrimaryKeys(arry);
                    }
                    recInfo.setValue(sri);
                }
            }
        }
        return recInfo;
    }
}

